package com.gis.aoitest.processfunction;

import com.gis.aoitest.beans.SensorReading;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class ProcessTest00_Timer {

    private static OutputTag<SensorReading> lowTempTag;
    public static void main(String[] args) throws Exception {
        // 执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读入数据源
        DataStreamSource<String> socketStream = env.socketTextStream("localhost", 9999);

        lowTempTag = new OutputTag<SensorReading>("lowTemp") {
        };

        // 处理
        SingleOutputStreamOperator<SensorReading> highStream = socketStream.map(data -> {
                    String[] fields = data.split(",");
                    return new SensorReading(fields[0], Long.parseLong(fields[1]), Double.parseDouble(fields[2]));
                })
                .keyBy("id")
                .process(new MySocketProcess());

        highStream.print("high-stream");
        highStream.getSideOutput(lowTempTag).print("low-stream");



//                .process(new KeyedProcessFunction<String, SensorReading, Object>() {
//
//
//                    @Override
//                    public void processElement(SensorReading value, KeyedProcessFunction<String, SensorReading, Object>.Context ctx, Collector<Object> out) throws Exception {
//                        long timer = ctx.timerService().currentProcessingTime();
//                        out.collect(value.getId() + "到达时间：" + new Timestamp(timer));
//
//                        // 注册一个十秒后的定时器
//                        ctx.timerService().registerProcessingTimeTimer(timer + 2 * 1000L);
//                        if (value.getId().equals("data")) {
//                            ctx.timerService().deleteProcessingTimeTimer(timer + 2 * 1000L);
//                            System.out.println(value.getId() + "定时器被删除");
//                        }
//                    }
//
//                    @Override
//                    public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, Object>.OnTimerContext ctx, Collector<Object> out) throws Exception {
//                        // 定时器触发
//                        out.collect(ctx.getCurrentKey() + "定时器触发：" + new Timestamp(ctx.timerService().currentProcessingTime()));
//                    }
//                }).print();

        // 执行关闭
        env.execute();


    }

    // 自定义类
    public static class MySocketProcess extends KeyedProcessFunction<Tuple, SensorReading, SensorReading>{


        private ValueState<Double> lastTemplate;
        private ValueState<Long> timerState;
        // 报警时间
        private ValueState<Long> alarmTime;

        @Override
        public void open(Configuration parameters) throws Exception {
            // 状态
            lastTemplate = getRuntimeContext().getState(new ValueStateDescriptor<Double>("last-temp", Double.class, Double.MIN_VALUE));
            timerState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("time-ts", Long.class));
            alarmTime = getRuntimeContext().getState(new ValueStateDescriptor<Long>("alarm-ts", Long.class, 0L));
        }

        @Override
        public void processElement(SensorReading value, KeyedProcessFunction<Tuple, SensorReading, SensorReading>.Context ctx, Collector<SensorReading> out) throws Exception {
            // 获取状态
            Double lastTemp = lastTemplate.value();
            Long timeTs = timerState.value();
            Long alarmTs = alarmTime.value();

            // 如果温度大于30,输入到highstream
            if (value.getTemperature() > 30) {
                out.collect(value);
            }else {
                ctx.output(lowTempTag, value);
            }

            // 如果温度增高，并且无定时器注册定时器
            if (value.getTemperature() > lastTemp && timeTs == null){
                long ts = ctx.timerService().currentProcessingTime() + 5 * 1000L;
                ctx.timerService().registerProcessingTimeTimer(ts);
                // 更新
                alarmTime.update(alarmTs + 5L);
                timerState.update(ts);
            }else if (value.getTemperature() < lastTemp && timeTs != null ){
                ctx.timerService().deleteProcessingTimeTimer(timeTs);
                alarmTime.update(0L);
                System.out.println("alarmTime: " + alarmTime.value() + "; alarmTs: " + alarmTs);
                timerState.clear();
            }else if (value.getTemperature() < lastTemp && timeTs == null) {
                alarmTime.update(0L);
                System.out.println("alarmTime: " + alarmTime.value() + "; alarmTs: " + alarmTs);
            }
            lastTemplate.update(value.getTemperature());
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<Tuple, SensorReading, SensorReading>.OnTimerContext ctx, Collector<SensorReading> out) throws Exception {
            System.out.println(ctx.getCurrentKey() + "连续" + alarmTime.value() + "温度告警");
            timerState.clear();
        }

        @Override
        public void close() throws Exception {
            System.out.println("close!!!");
            timerState.clear();
        }
    }

}
